package club.samubird.simpledatahandlekit.es.synckit;

import club.samubird.simpledatahandlekit.utils.BeanMapUtils;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.frameworkset.elasticsearch.boot.BBossESStarter;
import org.frameworkset.elasticsearch.client.ClientInterface;
import org.frameworkset.elasticsearch.client.ClientOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * @author sam
 * @date 2019/10/11 13:45
 * @description 从数据库同步到Es的核心类;提供核心的操作支持
 */
@Slf4j
@Component
public class SyncDb2EsCore {

	@Autowired
	private BBossESStarter bBossESStarter;

	@Autowired
	private JdbcTemplate jdbcTemplate;

	/**
	 * 对于 mysql 来说 一次查询 5000条 比较常见;
	 * 对于 es 来说一次查询 10000条 比较常见;
	 */
	private static final Integer BATCH_SIZE = 5000;

	private static final Long WAITE_TIME = 10000L;

	private ExecutorService executorService;

	@PostConstruct
	private void init() {
		executorService = new ThreadPoolExecutor(4, 8,
				0L, TimeUnit.MILLISECONDS,
				new LinkedBlockingQueue<Runnable>(50000), new ThreadFactoryBuilder().setNameFormat("sca_sync-从数据库同步到Es-%d").build());
	}

	private boolean checkAndPrepareConfig(SyncDb2EsOperationConfig config) {
		if (StringUtils.isBlank(config.getDbQueryWhereStr())) {
			config.setDbQueryWhereStr(" 1 = 1");
		}
		if (StringUtils.isBlank(config.getDbQueryColumnStr())) {
			config.setDbQueryColumnStr(" * ");
		}
		if (StringUtils.isBlank(config.getDbQueryOrderByStr())) {
			config.setDbQueryOrderByStr(config.getSourceTableKey());
		}
		return BeanMapUtils.hasEmptyPropertiesWithExcludeSome(config);
	}

	/**
	 * 根据配置文件进行同步
	 *
	 * @param config 同步配置
	 * @return 返回同步的数量
	 */
	public Integer syncDb2Es(SyncDb2EsOperationConfig config, SyncDb2EsDataConverter converter) {
		if (checkAndPrepareConfig(config)) {
			throw new RuntimeException("从 Db 同步到 Es 参数不合法, 参数是: " + config.toString());
		}

		String countSql = "SELECT COUNT(1) FROM " + config.getDbTableName() + " WHERE " + config.getDbQueryWhereStr();
		Integer total = jdbcTemplate.queryForObject(countSql, Integer.class);
		if (Objects.nonNull(total) && total > 0) {

			ClientOptions clientOptions = new ClientOptions();
			clientOptions.setIdField(config.getEsIdFiled());

			int num = (total + BATCH_SIZE - 1) / BATCH_SIZE;
			CountDownLatch countDownLatch = new CountDownLatch(num);
			int start = 0;
			int end = start;

			String querySql = "SELECT " + config.getDbQueryColumnStr() + " FROM " + config.getDbTableName() + " WHERE " + config.getDbQueryWhereStr() + " ORDER BY " + config.getDbQueryOrderByStr() + " ";
			do {
				end = total > end + BATCH_SIZE ? end + BATCH_SIZE : total;
				int finalStart = start;
				executorService.submit(
						() -> {
							String currentBatchQuerySql = querySql + " LIMIT " + finalStart + " , " + BATCH_SIZE;
							try {
								List<Map<String, Object>> currentQueryList = jdbcTemplate.queryForList(currentBatchQuerySql);
								if (CollectionUtils.isNotEmpty(currentQueryList)) {
									switch (config.getEsOptionType()) {
										case INSERT:
											syncInsert(currentQueryList, config, clientOptions, converter);
											break;
										case UPDATE:
											syncUpdate(currentQueryList, config, clientOptions, converter);
											break;
										case DELETE:
											syncDelete(currentQueryList, config, clientOptions, converter);
											break;
										default:
											break;
									}
								} else {
									log.info("根据 SQL {} 没有查到需要同步的数据", currentBatchQuerySql);
								}
							} catch (Exception e) {
								log.error("同步数据:  SQL {} 出现异常.", currentBatchQuerySql, e);
							} finally {
								countDownLatch.countDown();
							}
						}
				);
				start = end;
			} while (total > end);

			try {
				countDownLatch.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}

			try {
				// 一分钟后 手动执行强制刷新
				Thread.sleep(WAITE_TIME);
			} catch (Exception e) {
				log.error("Db 到 Es 同步操作 {}  出现异常: {}", config.getEsOptionType().name(), e);
			} finally {
				//  手动执行强制刷新
				ClientInterface restClient = bBossESStarter.getRestClient();
				Object result = restClient.executeRequest(config.getEsIndexName() + "/_refresh");
				log.info("Db 到 Es 同步操作 {} 完成后, 手动刷新结果是: {}", config.getEsOptionType().name(), result.toString());
			}
		} else {
			log.info("根据 SQL {} 统计到需要同步的数据数量为0", countSql);
		}
		return total;
	}

	private void syncDelete(List<Map<String, Object>> queryList, SyncDb2EsOperationConfig config, ClientOptions clientOptions, SyncDb2EsDataConverter converter) {
		ClientInterface restClient = bBossESStarter.getRestClient();
		String[] deleteIdArray = queryList.stream().map(m -> (String) m.get(config.getDbQueryColumnStr())).toArray(String[]::new);
		String deleteResponse = restClient.deleteDocuments(config.getEsIndexName(), config.getEsIndexType(), deleteIdArray);
		log.info(deleteResponse);
	}

	private void syncInsert(List<Map<String, Object>> queryList, SyncDb2EsOperationConfig config, ClientOptions clientOptions, SyncDb2EsDataConverter converter) {
		ClientInterface restClient = bBossESStarter.getRestClient();
		List insertList = converter.insertConverter(queryList);
		if (CollectionUtils.isNotEmpty(insertList)) {
			String addResponse = restClient.addDocuments(config.getEsIndexName(), config.getEsIndexType(), insertList, clientOptions);
			log.info(addResponse);
		} else {
			log.info("查询到的数据: {} 经过新增转换器 转换之后为空。", queryList);
		}
	}

	private void syncUpdate(List<Map<String, Object>> queryList, SyncDb2EsOperationConfig config, ClientOptions clientOptions, SyncDb2EsDataConverter converter) {
		ClientInterface restClient = bBossESStarter.getRestClient();
		List updateList = converter.updateConverter(queryList);
		if (CollectionUtils.isNotEmpty(updateList)) {
			String updateResponse = restClient.updateDocuments(config.getEsIndexName(), config.getEsIndexType(), updateList, clientOptions);
			log.info(updateResponse);
		} else {
			log.info("查询到的数据: {} 经过更新转换器 转换之后为空。", queryList);
		}
	}

}
